package mq

import (
	"context"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/inf/sarama"
	"github.com/pkg/errors"
)

type KafkaSender interface {
	SendMessage(ctx context.Context, topicName string, data string) (err error)
}

func NewKafkaSender(topicCluster map[string]string) *kafkaSender {
	bs := &kafkaSender{
		producers: make(map[string]sarama.SyncProducer),
	}
	config := sarama.NewConfig()
	var err error
	for topic, cluster := range topicCluster {
		bs.producers[topic], err = sarama.NewSyncProducerWithClusterName(cluster, config)
		if err != nil {
			panic(err)
		}
	}
	return bs

}

type kafkaSender struct {
	producers map[string]sarama.SyncProducer
}

func (b *kafkaSender) SendMessage(ctx context.Context, topic string, data string) (err error) {
	p, ok := b.producers[topic]
	if !ok {
		return biz_error.InvalidParam.Wrap(errors.Errorf("unknown topic, no defined cluster,topic:%s", topic))
	}

	msg := &sarama.ProducerMessage{Topic: topic, Value: sarama.StringEncoder(data)}

	var partition int32
	var offset int64

	partition, offset, err = p.CtxSendMessage(ctx, msg)
	if err != nil {
		log.V1.CtxError(ctx, "topic:%s, err:%v, msg:%s", topic, err, data)
		return err
	}
	log.V1.CtxInfo(ctx, "topic:%s, partition:%d, offset:%d, msg:%s", topic, partition, offset, data)

	return nil
}
